package com.zhanghe.study.spring4.beans.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;

/**
 * @author zh
 * @date 2023/1/3 15:36
 */
public class TestKafka {

    @Autowired
    private KafkaTemplate kafkaTemplate;



    public void send(){
        // 异步发送
        kafkaTemplate.send("topic","message").addCallback(new ListenableFutureCallback<SendResult<String,String>>(){

            @Override
            public void onSuccess(SendResult<String, String> result) {

            }

            @Override
            public void onFailure(Throwable ex) {

            }
        });

        // 同步发送
        ListenableFuture<SendResult<String,String>> future = kafkaTemplate.send("topic", "message");
        try {
            SendResult<String, String> sendResult = future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    @KafkaListener(topicPartitions = { // 设置主题分区
            @TopicPartition(topic = "testTopic",partitions = {"0","1"},partitionOffsets = {
                    @PartitionOffset(partition = "0",initialOffset = "10"),
                    @PartitionOffset(partition = "1",initialOffset = "100")
            })
    },
            concurrency = "10", // 并发度
            errorHandler = "myErrorHandler") // 失败处理，需要实现一个KafkaListenerErrorHandler接口
    public void accept(String msg){

    }

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public void test(){
        kafkaListenerEndpointRegistry.getListenerContainer("listenId").start();
        kafkaListenerEndpointRegistry.getListenerContainer("listenId").pause();
        kafkaListenerEndpointRegistry.getListenerContainer("listenId").resume();
        kafkaListenerEndpointRegistry.getListenerContainer("listenId").stop();
    }
}
